
/*
 * Copyright © 2021 https://www.cestc.cn/ All rights reserved.
 */

package com.zx.learn.flink.windows;

//import com.alibaba.fastjson.JSONObject;
//import lombok.AllArgsConstructor;
//import lombok.Data;
//import lombok.NoArgsConstructor;
//import org.apache.commons.lang3.StringUtils;
//import org.apache.commons.lang3.time.FastDateFormat;
//import org.apache.flink.api.common.RuntimeExecutionMode;
//import org.apache.flink.api.common.functions.AggregateFunction;
//import org.apache.flink.api.common.functions.FilterFunction;
//import org.apache.flink.api.common.functions.MapFunction;
//import org.apache.flink.api.java.tuple.Tuple2;
//import org.apache.flink.streaming.api.datastream.DataStream;
//import org.apache.flink.streaming.api.datastream.DataStreamSource;
//import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
//import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
//import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
//import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
//import org.apache.flink.streaming.api.windowing.time.Time;
//import org.apache.flink.streaming.api.windowing.triggers.ContinuousEventTimeTrigger;
//
//import java.util.HashSet;
//import java.util.Set;
//import java.util.concurrent.TimeUnit;

/**Java求日活的WindowFunction使用
 */
public class WindowFunctionDemo {
//    public static void main(String[] args) throws Exception {
//        //1.创建流环境
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
//        env.setParallelism(1);
//        //2.获取数据源
//        DataStreamSource<String> source = env.socketTextStream("localhost", 9999);
        // 设置时间语议，并过滤其中的首页曝光数据
//        DataStream<AppLogBean> homeExposureStream = appExposureStream
//                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<AppLogBean>(Time.seconds(0)) {
//                    @Override
//                    public long extractTimestamp(AppLogBean element) {
//                        return element.getTime() * 1000;
//                    }
//                })
//                .filter(new FilterFunction<AppLogBean>() {
//                    @Override
//                    public boolean filter(AppLogBean value) throws Exception {
//                        return "home_exposure".equals(value.getTopic()) && StringUtils.isNotBlank(value.getScdata());
//                    }
//                });
//
//// 获取出其中的用户id
//        SingleOutputStreamOperator<Tuple2<String, String>> userIdStream = homeExposureStream
//                .map(new MapFunction<AppLogBean, Tuple2<String, String>>() {
//                    @Override
//                    public Tuple2<String, String> map(AppLogBean appLogBean) throws Exception {
//
//                        String resultUserId = "1";
//
//                        JSONObject scdataJson = JSONObject.parseObject(appLogBean.getScdata());
//                        String user_id = scdataJson.getString("user_id");
//                        resultUserId = user_id;
//                        return Tuple2.of("dummy", resultUserId);
//                    }
//                });
//
//// 对用户id开窗，并统计每天的数据
//        SingleOutputStreamOperator<String> result = userIdStream
//                .keyBy(t -> t.f0)
//                .window(TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8)))
//                .trigger(ContinuousEventTimeTrigger.of(Time.seconds(1)))
//                .aggregate(new UniqueVisitorAggregateFunction(), new UniqueVisitorProcessWindowFunction());
//
//// 使用print打印数据
//        result.print("result>>>>>>>>>");
//
//    }
//
//    /**
//     * UV的窗口类
//     */
//    public static class UniqueVisitorProcessWindowFunction extends ProcessWindowFunction<Long, String, String, TimeWindow> {
//
//        private final FastDateFormat df = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");
//
//        @Override
//        public void process(String s, Context context, Iterable<Long> elements, Collector<String> out) throws Exception {
//            System.out.println("##### 当前的watermark为 #####" + df.format(context.currentWatermark()));
//            System.out.println("##### 窗口开始时间 ###########" + df.format(context.window().getStart()));
//            System.out.println("##### 窗口结束时间 ###########" + df.format(context.window().getEnd()));
//            System.out.println("##### 该窗口当前统计的UV #####" + elements.iterator().next());
//            out.collect("UV " + elements.iterator().next());
//        }
//    }
//
//    /**
//     * UV的聚合类
//     */
//    public static class UniqueVisitorAggregateFunction implements AggregateFunction<Tuple2<String, String>, Tuple2<Set<String>, Long>, Long> {
//
//        @Override
//        public Tuple2<Set<String>, Long> createAccumulator() {
//            return Tuple2.of(new HashSet<>(), 0L);
//        }
//
//        @Override
//        public Tuple2<Set<String>, Long> add(Tuple2<String, String> value, Tuple2<Set<String>, Long> accumulator) {
//            if (!accumulator.f0.contains(value.f1)) {
//                accumulator.f0.add(value.f1);
//                accumulator.f1 += 1;
//            }
//            return accumulator;
//        }
//
//        @Override
//        public Long getResult(Tuple2<Set<String>, Long> accumulator) {
//            return accumulator.f1;
//        }
//
//        @Override
//        public Tuple2<Set<String>, Long> merge(Tuple2<Set<String>, Long> a, Tuple2<Set<String>, Long> b) {
//            return null;
//        }
//    }

//    @Data
//    @AllArgsConstructor
//    @NoArgsConstructor
//    public static class AppLogBean {
//        private String sensorId;//信号灯id
//        private Integer count;//通过该信号灯的车的数量
//    }
}
